// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/kucoin_futures/feed_v1/processor.h"

#include <boost/algorithm/string/replace.hpp>
#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/kucoin/feed_v1/update.h"
#include "coin2/exchange/kucoin_futures/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::kucoin_futures::feed_v1::impl {

void KucoinFuturesFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("type")) {
    if (strcmp(document["type"].GetString(), "pong") == 0 ||
        strcmp(document["type"].GetString(), "welcome") == 0 ||
        strcmp(document["type"].GetString(), "ack") == 0) {
      return;
    }
    if (strcmp(document["type"].GetString(), "error") == 0) {
      LOG(INFO) << "[KucoinFutures Feed Processor] meta msg for " << packet.topic_name << ": "
                << packet.payload.data();
      return;
    }
  }

  if (!document.HasMember("subject") && !document.HasMember("data")) {
    LOG(INFO) << "[KucoinFutures Feed Processor] meta msg for "
              << packet.topic_name << ": " << packet.payload.data();
    return;
  }


  upd_mgr_.onBeginPacket(packet);
  if (document.HasMember("subject")) {
    auto channel = KucoinFuturesChannel::Parse(document["subject"].GetString());
    if (!CheckChannelFilter(channel.type)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (channel.type == "book") {  // book
      auto splited = Split(document["topic"].GetString(), ":");
      CHECK_EQ(splited.size(), 2);
      auto& channel_sub = splited[0];
      auto& native_symbol = splited[1];
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (channel_sub == "/contractMarket/level2Depth5" ||
            channel_sub == "/contractMarket/level2Depth50") {
          if (!stat.snapshot_received || !stat.update_received) {
            if (parser_config_.publish_bbo_book_only()) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (book_generated && is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else if (parser_config_.has_sampling_frequency_bps()) {
              bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
              bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
              if (parser_config_.has_trade_sampling_frequency_bps()) {
                auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                if (sampled ||
                    timespan > parser_config_.trade_sampling_time_length()) {
                  auto* trade = upd_mgr_.GetTradeMutable();
                  if (slim_trade->PopulateTradeAndClear(trade)) {
                    upd_mgr_.PublishTrade(native_symbol, false);
                  }
                }
              }
              if (book_generated && sampled) {
                upd_mgr_.PublishBook();
              }
            } else {
              bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
              if (book_generated) {
                upd_mgr_.PublishBook();
              }
            }
            stat.snapshot_received = true;
          }
        } else if (channel_sub == "/contractMarket/level2") {
          if (stat.snapshot_received || packet.timestamp < SNAPSHOT_ADDED_TS) {
            if (parser_config_.publish_bbo_book_only()) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              bool book_generated = KucoinFuturesFeedUpdateParser::ParseBookL2(
                  book, document["data"], true /* light bbo */);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (book_generated && is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else if (parser_config_.has_sampling_frequency_bps()) {
              bool book_generated = KucoinFuturesFeedUpdateParser::ParseBookL2(
                  book, document["data"], true /* light bbo */);
              bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
              if (parser_config_.has_trade_sampling_frequency_bps()) {
                auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
                int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
                if (sampled ||
                    timespan > parser_config_.trade_sampling_time_length()) {
                  auto* trade = upd_mgr_.GetTradeMutable();
                  if (slim_trade->PopulateTradeAndClear(trade)) {
                    upd_mgr_.PublishTrade(native_symbol, false);
                  }
                }
              }
              if (book_generated && sampled) {
                upd_mgr_.PublishBook();
              }
            } else {
              bool book_generated = KucoinFuturesFeedUpdateParser::ParseBookL2(
                  book, document["data"], false /* light bbo */);
              if (book_generated) {
                upd_mgr_.PublishBook();
              }
            }

            stat.update_received = true;
          }
        } else {
          LOG(ERROR) << "[KucoinFutures Feed Processor] unkown book: "
            << packet.payload.data();
        }
      }
    } else if (channel.type == "trade") {  // trade
      std::string native_symbol = document["data"]["symbol"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      Trade _trade{};
      int64_t timestamp = std::stoull(document["data"]["ts"].GetString());
      KucoinFeedUpdateParser::ParseTrade(document["data"], &_trade,
                                         timestamp);
      if (parser_config_.has_trade_sampling_frequency_bps()) {
        auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
            packet.timestamp,
            _trade.price,
            _trade.fill_qty,
            _trade.side,
            _trade.timestamp,
            parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      } else {
        auto* trade = upd_mgr_.GetTradeMutable();
        *trade = _trade;
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (channel.type == "instrument") {
      std::string native_symbol = document["topic"].GetString();
      boost::replace_all(native_symbol, "/contract/instrument:", "");

      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      int64_t ts =
          stoll(document["data"]["timestamp"].GetString()) * 1'000'000LL;
      KucoinFuturesFeedUpdateParser::ParseMarkIndexPrice(
          document["data"], ts, &upd_mgr_, native_symbol);
    }
  } else if (document.HasMember("url")) {
    auto url = std::string(document["url"].GetString());
    auto channel = KucoinFuturesChannel::Parse(url);
    if (!CheckChannelFilter(channel.type)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (document["data"].HasMember("code") &&
        strcmp(document["data"]["code"].GetString(), "200000") != 0) {
      LOG(INFO) << "[KucoinFutures Feed Processor] unkown topic: "
              << packet.payload.data();
      upd_mgr_.onEndPacket(packet);
      return;
    }
    if (channel.type == "open_interest") {
      for (const auto& contract : document["data"]["data"].GetArray()) {
        std::string native_symbol = contract["symbol"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          continue;
        }
        if (contract.HasMember("openInterest") && contract["openInterest"].IsString()) {
          auto* open_interest = upd_mgr_.GetOpenInterestMutable();
          open_interest->open_interest_qty = stod(contract["openInterest"].GetString());
          open_interest->timestamp = packet.timestamp;
          upd_mgr_.PublishOpenInterest(native_symbol);
        }
      }
    } else if (channel.type == "funding_rate") {
      std::string native_symbol =
          document["data"]["data"]["symbol"].GetString();
      boost::replace_all(native_symbol, ".", "");
      boost::replace_all(native_symbol, "FPI8H", "");
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      KucoinFuturesFeedUpdateParser::ParseFundingRate(
          document["data"]["data"], packet.timestamp, &upd_mgr_, native_symbol);
    } else if (channel.type == "kline") {
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      std::string native_symbol = parameters["symbol"];
      std::string kline_interval = parameters["granularity"];
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      auto interval = KucoinFuturesFeedUpdateParser::ParseKlineInterval(kline_interval);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        const auto& data = document["data"]["data"].GetArray();
        int page = -1;
        if (parameters.count("currentPage") > 0) {
          page = std::stoi(parameters["currentPage"]);
        }
        if (page == 0) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        bool to_publish = false;
        // for (int i = data.Size() - 1; i >= 0; i--) {
        for (auto& kline_data : data) {
          // auto& kline_data = data[i];
          bool freezed = KucoinFuturesFeedUpdateParser::ParseKlineRest(kline_data,
              klines, interval, packet.timestamp);
          to_publish = to_publish || freezed;
        }
        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    } else {
      LOG(ERROR) << "[KucoinFutures Feed Processor] unkown topic: "
              << packet.payload.data();
    }
  } else {
    LOG(ERROR) << "[KucoinFutures Feed Processor] unkown topic: "
            << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::kucoin_futures::feed_v1::impl
